package one.stand.service.impl;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import one.stand.code.RedisCode;
import one.stand.entity.TokenEntity;
import one.stand.entity.websocket.*;
import one.stand.enums.LoginEnum;
import one.stand.enums.MessageTypeEnum;
import one.stand.enums.SendTypeEnum;
import one.stand.model.ResultModel;
import one.stand.request.BusinessCardRequest;
import one.stand.request.ConsultRecordRequest;
import one.stand.response.BusinessCardResponse;
import one.stand.service.BaseService;
import one.stand.service.UserChannelService;
import one.stand.util.ChannelGroupUtil;
import one.stand.wesocket.code.CommandCode;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.Base64;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 */
@Service
public class UserChannelServiceImpl extends BaseService implements UserChannelService {
    /**
     * 定时器
     */
    private ScheduledExecutorService executorService;

    @Override
    public TokenEntity channelV(ChannelHandlerContext ctx, BaseSocketEntity baseSocketEntity) {
        TokenEntity tokenEntity = redisUtil.get(RedisCode.getUserLoginToken(baseSocketEntity.getToken()));
        if (null == tokenEntity) {
            ctx.close();
            return null;
        }
        if (baseSocketEntity.getCode().equals(CommandCode.SOCKET_REGISTER)) {
            return tokenEntity;
        }
        UserChannelEntity userChannelEntity = redisUtil.get(RedisCode.getUserWebSocket(tokenEntity.getId(), tokenEntity.getLoginProgramName()));
        if (null == userChannelEntity) {
            close(ctx);
            return null;
        }

        if (!redisUtil.hasKey(RedisCode.socketChanneKey(ctx.channel().id().asLongText()))) {
            log.info("通道错误：{}-{}", baseSocketEntity.toString(), ctx.channel().id().asLongText());
            close(ctx);
            return null;
        }
        return tokenEntity;
    }

    @Async("socketTask")
    @Override
    public void heartbeat(ChannelHandlerContext ctx, BaseSocketEntity baseSocketEntity) {
        executorService = ChannelGroupUtil.timerMap.get(ctx.channel().id().asLongText());
        if (null != executorService) {
            executorService.shutdown();
            ChannelGroupUtil.timerMap.remove(ctx.channel().id().asLongText());
        }
        // 移除之前心跳定时任务
        time(ctx, baseSocketEntity.getId(), baseSocketEntity.getLoginProgramName());
        //注册回复
        send(ctx, new SocketResultEntity() {{
            setCode(CommandCode.SOCKET_RESULT_HEARTBEAT);
            setVersion(baseSocketEntity.getVersion());
        }});

        GetUnreadMessageNumEntity getUnreadMessageNumEntity = new GetUnreadMessageNumEntity();
        getUnreadMessageNumEntity.setId(baseSocketEntity.getId());
        getUnreadMessageNumEntity.setLoginProgramName(baseSocketEntity.getLoginProgramName());
        getUnreadMessageNum(ctx, getUnreadMessageNumEntity);
    }

    @Override
    public void register(ChannelHandlerContext ctx, BaseSocketEntity baseSocketEntity) {
        UserChannelEntity userChannelEntity = new UserChannelEntity();
        userChannelEntity.setUserId(baseSocketEntity.getId());
        userChannelEntity.setChanneId(ctx.channel().id().asLongText());
        userChannelEntity.setRegister(currDate());
        userChannelEntity.setChannelIp(ctx.channel().remoteAddress().toString());
        redisUtil.set(RedisCode.getUserWebSocket(baseSocketEntity.getId(), baseSocketEntity.getLoginProgramName()), userChannelEntity);
        Channel channel = ctx.channel();
        ChannelGroupUtil.channelGroup.add(channel);
        ChannelGroupUtil.channelMap.put(channel.id().asLongText(), channel);


        // 移除之前心跳定时任务
        executorService = ChannelGroupUtil.timerMap.get(channel.id().asLongText());
        if (null != executorService) {
            executorService.shutdown();
            ChannelGroupUtil.timerMap.remove(channel.id().asLongText());
        }
        //开始心跳计时
        time(ctx, baseSocketEntity.getId(), baseSocketEntity.getLoginProgramName());


        //注册回复
        send(ctx, new SocketResultEntity() {{
            setCode(CommandCode.SOCKET_RESULT_REGISTER);
            setVersion(baseSocketEntity.getVersion());
        }});
        //缓存通道
        redisUtil.set(RedisCode.socketChanneKey(channel.id().asLongText()), baseSocketEntity.getId());

        GetUnreadMessageNumEntity getUnreadMessageNumEntity = new GetUnreadMessageNumEntity();
        getUnreadMessageNumEntity.setId(baseSocketEntity.getId());
        getUnreadMessageNumEntity.setLoginProgramName(baseSocketEntity.getLoginProgramName());
        getUnreadMessageNum(ctx, getUnreadMessageNumEntity);
    }

    @Async("socketTask")
    @Override
    public void sendMessage(ChannelHandlerContext ctx, SendMessageEntity sendMessageEntity) {
        if (sendMessageEntity.getTargetId().equals(sendMessageEntity.getId())) {
            log.info("不能给自己发送消息");
            return;
        }
        if (StringUtils.isEmpty(sendMessageEntity.getMessage())) {
            log.info("禁止发送空消息");
            return;
        }
        MessageEntity messageEntity = new MessageEntity();
        messageEntity.setMessageType(sendMessageEntity.getMessageType());
        messageEntity.setSendId(sendMessageEntity.getId());
        messageEntity.setTargetId(sendMessageEntity.getTargetId());
        messageEntity.setTime(sendMessageEntity.getTime());
        messageEntity.setMessage(sendMessageEntity.getMessage());
        messageEntity.setType(sendMessageEntity.getType());
        messageEntity.setContentType(sendMessageEntity.getContentType());
        //消息存储
        redisMessage(messageEntity, sendMessageEntity.getLoginProgramName());
        /**
         * 添加咨询记录
         */
        addMessageDetail(sendMessageEntity);


        //推送消息给指定用户
        UserChannelEntity userChannelEntity = redisUtil.get(RedisCode.getUserWebSocket(sendMessageEntity.getTargetId(), getTargetLoginProgramName(sendMessageEntity.getLoginProgramName())));
        sendMessage:
        if (userChannelEntity != null) {
            Channel channel = ChannelGroupUtil.channelMap.get(userChannelEntity.getChanneId());
            if (null != channel) {
                //推送消息给目标
                sendMessageEntity.setSendId(sendMessageEntity.getId());
                sendMessageEntity.setCode(CommandCode.SOCKET_ACTIVE_SEND_MESSAGE);
                send(channel, sendMessageEntity);
            } else {
                //用户通道错误进行服务通知
                push(sendMessageEntity);
            }
        } else {
            //用户不在线进行服务通知
            push(sendMessageEntity);
        }


        sendMessageEntity.setCode(CommandCode.SOCKET_RESULT_SENDMESSAGE);
        sendMessageEntity.setSendId(sendMessageEntity.getId());
        sendMessageEntity.setType(messageEntity.getType());
        send(ctx, sendMessageEntity);

    }

    @Async("socketTask")
    @Override
    public void seeMessage(ChannelHandlerContext ctx, SeeMessageEntity seeMessageEntity) {
        log.info("读去消息：{}", JSON.toJSONString(seeMessageEntity));
        MessageEntity messageEntity = redisUtil.rGetList(RedisCode.getUserWebSocketMessageRecord(seeMessageEntity.getId(), seeMessageEntity.getTargetId(), seeMessageEntity.getLoginProgramName()));

        if (null == messageEntity || messageEntity.getSendId().equals(seeMessageEntity.getId())) {
            return;
        }

        //获取发送人的通道信息
        UserChannelEntity userChannelEntity = redisUtil.get(RedisCode.getUserWebSocket(seeMessageEntity.getTargetId(), getTargetLoginProgramName(seeMessageEntity.getLoginProgramName())));

        //未读数量
        String unreadNumKey = RedisCode.getUnreadNum(seeMessageEntity.getId().toString(), seeMessageEntity.getTargetId().toString(), seeMessageEntity.getLoginProgramName());
        redisUtil.remove(unreadNumKey);

        //发送者
        UnreadMessageEntity sendU = redisUtil.get(RedisCode.getUnreadMessage(seeMessageEntity.getId(), seeMessageEntity.getTargetId(), seeMessageEntity.getLoginProgramName()));
        if (sendU != null) {
            Long sendStartIndex = sendU.getStartIndex();
            Long sendEndIndex = sendU.getSize() + sendU.getStartIndex();
            //发送者消息模块
            List<MessageEntity> targetList = redisUtil.listRange(RedisCode.getUserWebSocketMessageRecord(seeMessageEntity.getId(), seeMessageEntity.getTargetId(), seeMessageEntity.getLoginProgramName()), sendStartIndex, sendEndIndex);
            //删除未查看的记录
            redisUtil.remove(RedisCode.getUnreadMessage(seeMessageEntity.getId(), seeMessageEntity.getTargetId(), seeMessageEntity.getLoginProgramName()));
            //修改状态为已读
            upd(targetList, sendStartIndex, seeMessageEntity.getId(), seeMessageEntity.getTargetId(), seeMessageEntity.getLoginProgramName());

            //如果用户在线那么就推送消息自己的信息被查看
            if (userChannelEntity != null) {
                //获取用户通道
                Channel channel = ChannelGroupUtil.channelMap.get(userChannelEntity.getChanneId());
                if (channel != null) {
                    //通知用户自己的消息被查看
                    seeMessageEntity.setCode(CommandCode.SOCKET_ACTIVE__SEE);
                    send(channel, seeMessageEntity);
                    log.info("读取消息，目标下线不推送消息");
                }

            }

        }

        //接收者
        UnreadMessageEntity receiveU = redisUtil.get(RedisCode.getUnreadMessage(seeMessageEntity.getTargetId(), seeMessageEntity.getId(), getTargetLoginProgramName(seeMessageEntity.getLoginProgramName())));
        if (receiveU != null) {
            Long receiveStartIndex = receiveU.getStartIndex();
            Long receiveEndIndex = receiveU.getSize() + receiveU.getStartIndex();
            List<MessageEntity> myList = redisUtil.listRange(RedisCode.getUserWebSocketMessageRecord(seeMessageEntity.getTargetId(), seeMessageEntity.getId(), getTargetLoginProgramName(seeMessageEntity.getLoginProgramName())), receiveStartIndex, receiveEndIndex);
            //删除未查看的记录
            redisUtil.remove(RedisCode.getUnreadMessage(seeMessageEntity.getTargetId(), seeMessageEntity.getId(), getTargetLoginProgramName(seeMessageEntity.getLoginProgramName())));
            //修改状态为已读
            upd(myList, receiveStartIndex, seeMessageEntity.getTargetId(), seeMessageEntity.getId(), getTargetLoginProgramName(seeMessageEntity.getLoginProgramName()));
        }
        send(ctx, new SocketResultEntity() {{
            setCode(CommandCode.SOCKET_RESULT_SEE);
            setVersion(seeMessageEntity.getVersion());
        }});

        GetUnreadMessageNumEntity getUnreadMessageNumEntity = new GetUnreadMessageNumEntity();
        getUnreadMessageNumEntity.setId(seeMessageEntity.getId());
        getUnreadMessageNumEntity.setLoginProgramName(seeMessageEntity.getLoginProgramName());
        getUnreadMessageNum(ctx, getUnreadMessageNumEntity);

    }

    @Async("socketTask")
    @Override
    public void getMessage(ChannelHandlerContext ctx, GetMessageEntity getMessageEntity) {
        if (getMessageEntity.getTargetId() == null) {
            return;
        }
        if (getMessageEntity.getTargetId().equals(getMessageEntity.getId())) {
            log.info("不能获取和自己的聊天记录。。。");
            return;
        }
        Long size = redisUtil.listSize(RedisCode.getUserWebSocketMessageRecord(getMessageEntity.getId(), getMessageEntity.getTargetId(), getMessageEntity.getLoginProgramName()));
        //读取消息列表进行消息查看操作
        SeeMessageEntity seeMessageEntity = new SeeMessageEntity();
        BeanUtils.copyProperties(getMessageEntity, seeMessageEntity);
        seeMessage(ctx, seeMessageEntity);
        lm:
        if (size > 0) {
            if (getMessageEntity.getClientNum() == null) {
                getMessageEntity.setClientNum(0L);
            }
            if (size.equals(getMessageEntity.getClientNum())) {
                log.info("数据到头了");
                break lm;
            }
            Long start;
            if (getMessageEntity.getClientNum() == 0) {
                start = size - 20;
            } else {
                start = size - getMessageEntity.getClientNum() - getMessageEntity.getPageSize();
            }
            Long end = start + getMessageEntity.getPageSize();
            if (start < 0) {
                start = 0L;
            }
            List<MessageEntity> messageEntities = redisUtil.listRange(RedisCode.getUserWebSocketMessageRecord(getMessageEntity.getId(), getMessageEntity.getTargetId(), getMessageEntity.getLoginProgramName()), start, end);
            getMessageEntity.setData(messageEntities);

        }
        //推送聊天列表
        getMessageEntity.setCode(CommandCode.SOCKET_RESULT_GET_MESSAGE);
        send(ctx, getMessageEntity);

        //新的推送
        if (getMessageEntity.getLoginProgramName().equals(LoginEnum.USER.getCode())) {

            //如果为空 那么就是第一次进入  推送初始消息
            company:
            if (size == 0) {
                Long cardId = null;
                if (!StringUtils.isEmpty(getMessageEntity.getCardId()) && !"undefined".equals(getMessageEntity.getCardId())) {
                    try {
                        cardId = Long.parseLong(getMessageEntity.getCardId());
                    } catch (Exception e) {
                        log.info("cardId解析失败:{}", e);
                    }
                }
                if (cardId == null) {
                    break company;
                }
                CompanyInfoEntity companyInfoEntity = crmFegin.messageGetCompanyInfo(cardId).getData();
                if (null == companyInfoEntity) {
                    break company;
                }
                log.info("获取公司消息：{}", JSON.toJSONString(companyInfoEntity));

                SendMessageEntity sendMessageEntity = new SendMessageEntity();
                BeanUtils.copyProperties(getMessageEntity, sendMessageEntity);
                sendMessageEntity.setType(1);
                sendMessageEntity.setMessageType(MessageTypeEnum.INITIAL_NEWS.getType());
                sendMessageEntity.setId(getMessageEntity.getTargetId());
                sendMessageEntity.setTargetId(getMessageEntity.getId());
                sendMessageEntity.setMessage(companyInfoEntity);
                sendMessageEntity.setTime(getMessageEntity.getTime());
                sendMessageEntity.setLoginProgramName(getTargetLoginProgramName(getMessageEntity.getLoginProgramName()));
                //消息推送
                sendMessage(ctx, sendMessageEntity);
            }

        }

        //如果是从商品进入的那么就进行商品消息推送
        commodity:
        if (getMessageEntity.getIntoType() == 1) {
            CommodityInfoEntity commodityInfoEntity = crmFegin.getSendMessage(getMessageEntity.getSpecId()).getData();
            if (null == commodityInfoEntity) {
                break commodity;
            }
            log.info("获取商品消息：{}", JSON.toJSONString(commodityInfoEntity));
            SendMessageEntity sendMessageEntity = new SendMessageEntity();
            BeanUtils.copyProperties(getMessageEntity, sendMessageEntity);
            sendMessageEntity.setMessageType(MessageTypeEnum.GOODS_NEWS.getType());
            sendMessageEntity.setCode(CommandCode.SOCKET_ACTIVE_SEND_MESSAGE);
            sendMessageEntity.setId(getMessageEntity.getId());
            sendMessageEntity.setTargetId(getMessageEntity.getTargetId());
            sendMessageEntity.setMessage(commodityInfoEntity);
            sendMessageEntity.setTime(getMessageEntity.getTime());
            //消息推送
            sendMessage(ctx, sendMessageEntity);
        }

        GetUnreadMessageNumEntity getUnreadMessageNumEntity = new GetUnreadMessageNumEntity();
        getUnreadMessageNumEntity.setId(getMessageEntity.getId());
        getUnreadMessageNumEntity.setLoginProgramName(getMessageEntity.getLoginProgramName());
        getUnreadMessageNum(ctx, getUnreadMessageNumEntity);
    }

    @Async("socketTask")
    @Override
    public void dynamicReview(ChannelHandlerContext ctx, DynamicReviewEntity dynamicReviewEntity) {


        UserChannelEntity userChannelEntity = redisUtil.get(RedisCode.getUserWebSocket(dynamicReviewEntity.getTargetId(), getTargetLoginProgramName(dynamicReviewEntity.getLoginProgramName())));
        if (userChannelEntity != null) {
            Channel channel = ChannelGroupUtil.channelMap.get(userChannelEntity.getChanneId());
            if (null != channel) {
                send(channel, new SocketResultEntity() {{
                    setCode(CommandCode.SOCKET_ACTIVE__DYNAMIC_REVIEW);
                    setVersion(dynamicReviewEntity.getVersion());
                }});
            }
        }


        //回复通知成功
        dynamicReviewEntity.setCode(CommandCode.SOCKET_RESULT_DYNAMIC_REVIEW);
        send(ctx, dynamicReviewEntity);

    }

    @Async("socketTask")
    @Override
    public void dynamicFabulous(ChannelHandlerContext ctx, DynamicFabulousEntity dynamicFabulousEntity) {
        send(ctx, new SocketResultEntity() {{
            setCode(CommandCode.SOCKET_ACTIVE__DYNAMIC_FABULOUS);
            setVersion(dynamicFabulousEntity.getVersion());
        }});

        dynamicFabulousEntity.setCode(CommandCode.SOCKET_RESULT_DYNAMIC_FABULOUS);
        send(ctx, dynamicFabulousEntity);

    }

    @Async("socketTask")
    @Override
    public void getImList(ChannelHandlerContext ctx, GetImListEntity getImListEntity) {
        Set<Integer> userIds = redisUtil.getZSetOrder(RedisCode.getImList(getImListEntity.getId(), getImListEntity.getLoginProgramName()));
        Integer[] ids = new Integer[userIds.size()];
        userIds.toArray(ids);
        ResultModel<List<BusinessCardResponse>> resultModel = userFeign.getUserInfo(new BusinessCardRequest() {{
            setUserIds(ids);
        }});
        List<ImListEntity> imList = new LinkedList<>();
        if (resultModel.getData() != null && resultModel.getData().size() > 0) {
            resultModel.getData().forEach(r -> {
                ImListEntity entity = new ImListEntity();
                entity.setLogo(r.getAvatarUrl());
                entity.setName(StringUtils.isEmpty(r.getNickeName()) ? "未知" : new String(Base64.getDecoder().decode(r.getNickeName().getBytes())));
                entity.setUserId(r.getUserId().longValue());
                entity.setNewestMessage(redisUtil.rGetList(RedisCode.getUserWebSocketMessageRecord(getImListEntity.getId(), r.getUserId().longValue(), getImListEntity.getLoginProgramName())));
                entity.setCardId(r.getCardId());
                entity.setPhone(r.getWxPhone());
                entity.setWxCode(r.getPersonalWx());
                imList.add(entity);
            });
        }
        getImListEntity.setImList(imList);
        getImListEntity.setCode(CommandCode.SOCKET_RESULT_IMLIST);
        send(ctx, getImListEntity);


        GetUnreadMessageNumEntity getUnreadMessageNumEntity = new GetUnreadMessageNumEntity();
        getUnreadMessageNumEntity.setId(getImListEntity.getId());
        getUnreadMessageNumEntity.setLoginProgramName(getImListEntity.getLoginProgramName());
        getUnreadMessageNum(ctx, getUnreadMessageNumEntity);
    }

    @Async("socketTask")
    @Override
    public void getUnreadMessageNum(ChannelHandlerContext ctx, GetUnreadMessageNumEntity getUnreadMessageNumEntity) {

        if (getUnreadMessageNumEntity.getTargetId() != null) {
            UnreadMessageEntity unreadMessageEntity = new UnreadMessageEntity();
            //未读数量
            Integer num = redisUtil.get(RedisCode.getUnreadNum(getUnreadMessageNumEntity.getId().toString(), getUnreadMessageNumEntity.getTargetId().toString(), getUnreadMessageNumEntity.getLoginProgramName()));
            if (num == null) {
                num = 0;
            }
            unreadMessageEntity.setSize(num);
            log.info("获取与目标的未读消息数量:{}", JSON.toJSONString(unreadMessageEntity));
            getUnreadMessageNumEntity.setUnreadMessageEntity(unreadMessageEntity);
            getUnreadMessageNumEntity.setCode(CommandCode.SOCKET_RESULT_UNREAD_MESSAGE_NUM);
            send(ctx, getUnreadMessageNumEntity);
        } else {
            Set<String> keys = redisUtil.getKey(RedisCode.getUnreadNum(getUnreadMessageNumEntity.getId().toString(), "*", getUnreadMessageNumEntity.getLoginProgramName()));
            log.info("获取全部未读消息：{}", keys.toString());
            UnreadMessageEntity unreadMessageEntity = new UnreadMessageEntity();
            unreadMessageEntity.setSize(0);
            for (String v : keys) {
                Integer num = redisUtil.get(v);
                if (num == null) {
                    num = 0;
                }
                unreadMessageEntity.setSize(unreadMessageEntity.getSize() + num);
            }
            getUnreadMessageNumEntity.setUnreadMessageEntity(unreadMessageEntity);
            getUnreadMessageNumEntity.setCode(CommandCode.SOCKET_RESULT_UNREAD_MESSAGE_NUM);
            send(ctx, getUnreadMessageNumEntity);
        }

    }

    /**
     * 获取目标登陆平台名
     *
     * @param loginProgramName
     * @return
     */
    public String getTargetLoginProgramName(String loginProgramName) {
        if (LoginEnum.USER.getCode().equals(loginProgramName)) {
            return LoginEnum.CRM.getCode();
        }
        if (LoginEnum.CRM.getCode().equals(loginProgramName)) {
            return LoginEnum.USER.getCode();
        }

        return null;
    }

    public void upd(List<MessageEntity> myList, Long startIndex, Long id, Long id2, String loginProgramName) {
        if (null != myList) {
            //进行跟新操作 让消息态改为已经查看
            for (int i = 0; i < myList.size(); i++) {
                MessageEntity receiveM = myList.get(i);
                receiveM.setType(1);
                redisUtil.rSet(RedisCode.getUserWebSocketMessageRecord(id, id2, loginProgramName), receiveM, startIndex + i);
            }
        }
    }

    /**
     * @param messageEntity
     * @param loginProgramName 发送者的登陆平台名
     */
    public void redisMessage(MessageEntity messageEntity, String loginProgramName) {
        //发送者
        Long sendIndex = redisUtil.rAddList(RedisCode.getUserWebSocketMessageRecord(messageEntity.getSendId(), messageEntity.getTargetId(), loginProgramName), messageEntity);
        //接收者
        Long receiveIndex = redisUtil.rAddList(RedisCode.getUserWebSocketMessageRecord(messageEntity.getTargetId(), messageEntity.getSendId(), getTargetLoginProgramName(loginProgramName)), messageEntity);


        //发送者
        UnreadMessageEntity sendU = redisUtil.get(RedisCode.getUnreadMessage(messageEntity.getSendId(), messageEntity.getTargetId(), loginProgramName));

        if (null == sendU) {
            sendU = new UnreadMessageEntity();
            sendU.setSize(1);
            sendU.setStartIndex(sendIndex - 1);
        } else {
            sendU.setSize(sendU.getSize() + 1);
        }


        //接收者
        UnreadMessageEntity receiveU = redisUtil.get(RedisCode.getUnreadMessage(messageEntity.getTargetId(), messageEntity.getSendId(), getTargetLoginProgramName(loginProgramName)));
        if (null == receiveU) {
            receiveU = new UnreadMessageEntity();
            receiveU.setSize(1);
            receiveU.setStartIndex(receiveIndex - 1);
        } else {
            receiveU.setSize(receiveU.getSize() + 1);
        }
        //发送者
        redisUtil.set(RedisCode.getUnreadMessage(messageEntity.getSendId(), messageEntity.getTargetId(), loginProgramName), sendU);
        //接收者
        redisUtil.set(RedisCode.getUnreadMessage(messageEntity.getTargetId(), messageEntity.getSendId(), getTargetLoginProgramName(loginProgramName)), receiveU);

        //建立好友关系
        redisUtil.addZSet(RedisCode.getImList(messageEntity.getSendId(), loginProgramName), messageEntity.getTargetId());

        redisUtil.addZSet(RedisCode.getImList(messageEntity.getTargetId(), getTargetLoginProgramName(loginProgramName)), messageEntity.getSendId());

        if (0 == messageEntity.getType()) {
            String unreadNumKey = RedisCode.getUnreadNum(messageEntity.getTargetId().toString(), messageEntity.getSendId().toString(), getTargetLoginProgramName(loginProgramName));
            Integer num = redisUtil.get(unreadNumKey);
            if (num == null) {
                num = 1;
            } else {
                num = num + 1;
            }
            redisUtil.set(unreadNumKey, num);
        }

    }

    @Async("socketTask")
    public void addMessageDetail(SendMessageEntity sendMessageEntity) {
        /**
         * 添加咨询记录
         */
        if (sendMessageEntity.getLoginProgramName().equals(LoginEnum.USER.getCode())) {
            ConsultRecordRequest recordRequest = new ConsultRecordRequest();
            recordRequest.setSonsultType(sendMessageEntity.getMessageType());
            recordRequest.setId(sendMessageEntity.getId());
            List<BusinessCardResponse> response = userFeign.getUserInfo(new BusinessCardRequest() {{
                Integer[] ids = {sendMessageEntity.getTargetId().intValue()};
                setUserIds(ids);
            }}).getData();
            log.info("内容返回：{}", JSON.toJSONString(response));
            if (null != response && response.size() >= 0) {
                recordRequest.setCardId(response.get(0).getCardId());
                recordRequest.setCompanyId(response.get(0).getCompanyId());
                ResultModel resultModel = userFeign.addConsultRecord(recordRequest);
                log.info("咨询记录返回：{}", JSON.toJSONString(resultModel));
            }

        }
    }

    public void push(SendMessageEntity sendMessageEntity) {
        //用户不在线则进行服务通知
        String text = null;
        if (sendMessageEntity.getMessageType().equals(MessageTypeEnum.GENERAL_NEWS.getType())) {
            text = sendMessageEntity.getMessage().toString();
        }
        if (sendMessageEntity.getMessageType().equals(MessageTypeEnum.GOODS_NEWS.getType())) {
            CommodityInfoEntity commodityInfoEntity = (CommodityInfoEntity) sendMessageEntity.getMessage();
            text = commodityInfoEntity.getGoodsName();
        }
        if (sendMessageEntity.getMessageType().equals(MessageTypeEnum.INITIAL_NEWS.getType())) {
            CompanyInfoEntity companyInfoEntity = (CompanyInfoEntity) sendMessageEntity.getMessage();
            text = companyInfoEntity.getCompanyName();
        }
        if (StringUtils.isEmpty(text)) {
            return;
        }
        //1文字消息，2图片消息，3语音消息，4视频消息
        switch (sendMessageEntity.getContentType()) {
            case 2:
                text = "你有一条未读图片";
                break;
            case 3:
                text = "你有一条未读语音";
                break;
            case 4:
                text = "你有一条未读视频";
                break;
        }
        if (sendMessageEntity.getLoginProgramName().equals(LoginEnum.USER.getCode())) {

            //当前消息来源是crm就给用户端发送服务通知
            ResultModel resultModel = userFeign.unreadMessage(sendMessageEntity.getId().intValue(), sendMessageEntity.getTargetId().intValue(), text, SendTypeEnum.UNREAD_MESSAGE.getCode());
            log.info("CRM端未读消息推送：{}", JSON.toJSON(resultModel));
        }
        if (sendMessageEntity.getLoginProgramName().equals(LoginEnum.CRM.getCode())) {
            //当前消息来源是用户端就给User端发送服务通知
            ResultModel resultModel = crmFegin.unreadMessage(sendMessageEntity.getId().intValue(), sendMessageEntity.getTargetId().intValue(), text, SendTypeEnum.UNREAD_MESSAGE.getCode());
            log.info("用户端未读消息推送：{}", JSON.toJSON(resultModel));
        }
    }

    /**
     * 心调超时逻辑
     *
     * @param ctx
     * @param id
     */
    public void time(final ChannelHandlerContext ctx, final Long id, final String loginProgramName) {
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());
        executorService.scheduleAtFixedRate(new Runnable() {
            private int num = 0;

            @Override
            public void run() {
                num++;
                if (num >= 2) {
                    log.info("心跳超时用户下线：" + id);
                    close(ctx);
                    redisUtil.remove(RedisCode.socketChanneKey(ctx.channel().id().asLongText()));
                    ChannelGroupUtil.timerMap.remove(ctx.channel().id().asLongText());
                    executorService.shutdown();

                }
            }
        }, 5, 5, TimeUnit.SECONDS);
        ChannelGroupUtil.timerMap.put(ctx.channel().id().asLongText(), executorService);
    }
}
